package com.huan.outputformat;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

public class FRecordWriter extends RecordWriter<Text, NullWritable> {


    FSDataOutputStream fosatguigu;
    FSDataOutputStream fosother;

    public FRecordWriter(TaskAttemptContext job) {

        try {
            //TODO 获取fs文件
            FileSystem fs = FileSystem.get( job.getConfiguration() );
            //TODO 创建atguigu.log输出流
            fosatguigu = fs.create( new Path( "E:\\atguigu.log" ) );
            //TODO 创建orter.log输出流
            fosother = fs.create( new Path( "E:\\other.log" ) );

        } catch (IOException e) {
            e.printStackTrace();
        }


    }

    @Override
    public void write(Text key, NullWritable value) throws IOException, InterruptedException {
        //TODO 判断key当中是否含有atguigu,如果有写出atguigu.log  , 如果没有写出到other.log
        if (key.toString().contains( "atguigu" )) {
            fosatguigu.write( key.toString().getBytes() );
        } else {
            fosother.write( key.toString().getBytes() );
        }
    }

    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {

        IOUtils.closeStream( fosatguigu );
        IOUtils.closeStream( fosother );

    }
}
